#include <Disks/DiskObjectStorage/ObjectStorages/ObjectStorageFactory.h>

#if USE_AWS_S3
#include <Disks/DiskObjectStorage/ObjectStorages/S3/DiskS3Utils.h>
#include <Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h>
#include <Disks/DiskObjectStorage/ObjectStorages/S3/diskSettings.h>
#endif

#if USE_HDFS
#include <Disks/DiskObjectStorage/ObjectStorages/HDFS/HDFSObjectStorage.h>
#include <Storages/ObjectStorage/HDFS/HDFSCommon.h>
#endif

#if USE_AZURE_BLOB_STORAGE
#include <Disks/DiskObjectStorage/ObjectStorages/AzureBlobStorage/AzureObjectStorage.h>
#include <Disks/DiskObjectStorage/ObjectStorages/AzureBlobStorage/AzureBlobStorageCommon.h>
#endif

#include <Disks/DiskObjectStorage/ObjectStorages/Web/WebObjectStorage.h>
#include <Disks/DiskObjectStorage/ObjectStorages/Local/LocalObjectStorage.h>
#include <Disks/loadLocalDiskConfig.h>
#include <Disks/DiskType.h>

#include <Interpreters/Context.h>

#include <Common/Macros.h>

#include <Core/Settings.h>

#include <filesystem>
#include <utility>

namespace fs = std::filesystem;

namespace DB
{
namespace Setting
{
    extern const SettingsUInt64 hdfs_replication;
}

namespace ErrorCodes
{
    extern const int NO_ELEMENTS_IN_CONFIG;
    extern const int UNKNOWN_ELEMENT_IN_CONFIG;
    extern const int BAD_ARGUMENTS;
    extern const int LOGICAL_ERROR;
}

ObjectStorageFactory & ObjectStorageFactory::instance()
{
    static ObjectStorageFactory factory;
    return factory;
}

void ObjectStorageFactory::registerObjectStorageType(const std::string & type, Creator creator)
{
    if (!registry.emplace(type, creator).second)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "ObjectStorageFactory: the metadata type '{}' is not unique", type);
}

ObjectStoragePtr ObjectStorageFactory::create(
    const std::string & name,
    const Poco::Util::AbstractConfiguration & config,
    const std::string & config_prefix,
    const ContextPtr & context,
    bool skip_access_check) const
{
    std::string type;
    if (config.has(config_prefix + ".object_storage_type"))
        type = config.getString(config_prefix + ".object_storage_type");
    else if (config.has(config_prefix + ".type")) /// .type -- for compatibility.
        type = config.getString(config_prefix + ".type");
    else
    {
        throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG, "Expected `object_storage_type` in config");
    }

    const auto it = registry.find(type);

    if (it == registry.end())
    {
        throw Exception(ErrorCodes::UNKNOWN_ELEMENT_IN_CONFIG,
                        "ObjectStorageFactory: unknown object storage type: {}", type);
    }

    return it->second(name, config, config_prefix, context, skip_access_check);
}

#if USE_AWS_S3
namespace
{

S3::URI getS3URI(const Poco::Util::AbstractConfiguration & config, const std::string & config_prefix, const ContextPtr & context)
{
    String endpoint = context->getMacros()->expand(config.getString(config_prefix + ".endpoint"));
    String endpoint_subpath;
    if (config.has(config_prefix + ".endpoint_subpath"))
        endpoint_subpath = context->getMacros()->expand(config.getString(config_prefix + ".endpoint_subpath"));

    S3::URI uri(fs::path(endpoint) / endpoint_subpath);

    /// An empty key remains empty.
    if (!uri.key.empty() && !uri.key.ends_with('/'))
        uri.key.push_back('/');

    return uri;
}

}

static std::string getEndpoint(
        const Poco::Util::AbstractConfiguration & config,
        const std::string & config_prefix,
        const ContextPtr & context)
{
    return context->getMacros()->expand(config.getString(config_prefix + ".endpoint"));
}

void registerS3ObjectStorage(ObjectStorageFactory & factory)
{
     auto creator = [](
        const std::string & name,
        const Poco::Util::AbstractConfiguration & config,
        const std::string & config_prefix,
        const ContextPtr & context,
        bool /* skip_access_check */) -> ObjectStoragePtr
    {
        auto uri = getS3URI(config, config_prefix, context);
        auto s3_capabilities = getCapabilitiesFromConfig(config, config_prefix);
        auto endpoint = getEndpoint(config, config_prefix, context);
        auto settings = std::make_unique<S3Settings>();
        settings->loadFromConfigForObjectStorage(config, config_prefix, context->getSettingsRef(), uri.uri.getScheme(), true);
        auto client = getClient(endpoint, *settings, context, /* for_disk_s3 */ true, name);
        auto key_generator = getKeyGenerator(uri, config, config_prefix);

        return std::make_shared<S3ObjectStorage>(std::move(client), std::move(settings), uri, s3_capabilities, key_generator, name);
    };

    factory.registerObjectStorageType("s3", creator);
    factory.registerObjectStorageType("s3_with_keeper", creator);
    factory.registerObjectStorageType("s3_plain", creator);
    factory.registerObjectStorageType("s3_plain_rewritable", creator);
}

#endif

#if USE_HDFS
void registerHDFSObjectStorage(ObjectStorageFactory & factory)
{
    factory.registerObjectStorageType(
        "hdfs",
        [](const std::string & /* name */,
           const Poco::Util::AbstractConfiguration & config,
           const std::string & config_prefix,
           const ContextPtr & context,
           bool /* skip_access_check */) -> ObjectStoragePtr
        {
            auto uri = context->getMacros()->expand(config.getString(config_prefix + ".endpoint"));
            checkHDFSURL(uri);
            if (uri.back() != '/')
                throw Exception(ErrorCodes::BAD_ARGUMENTS, "HDFS path must ends with '/', but '{}' doesn't.", uri);

            std::unique_ptr<HDFSObjectStorageSettings> settings = std::make_unique<HDFSObjectStorageSettings>(
                config.getUInt64(config_prefix + ".min_bytes_for_seek", 1024 * 1024), context->getSettingsRef()[Setting::hdfs_replication]);

            return std::make_shared<HDFSObjectStorage>(uri, std::move(settings), config, /* lazy_initialize */false);
        });
}
#endif

#if USE_AZURE_BLOB_STORAGE
void registerAzureObjectStorage(ObjectStorageFactory & factory)
{
    auto creator = [](
        const std::string & name,
        const Poco::Util::AbstractConfiguration & config,
        const std::string & config_prefix,
        const ContextPtr & context,
        bool /* skip_access_check */) -> ObjectStoragePtr
    {
        auto azure_settings = AzureBlobStorage::getRequestSettings(config, config_prefix, context->getSettingsRef());

        /// AzureObjectStorage::getCommonKeyPrefix() was not implemented previousely by mistake and was always returning an empty string.
        /// However, we use this string as ZooKeeper path for disk with metadata in Keeper.
        /// So, all instances of azure-with-keeper were using the same (empty, or root) path.
        /// We keep using empty prefix by default for compatibility, but allow to configure another one
        const String & common_key_prefix = config.getString(config_prefix + ".common_key_prefix_for_azure", "");

        AzureBlobStorage::ConnectionParams params
        {
            .endpoint = AzureBlobStorage::processEndpoint(config, config_prefix),
            .auth_method = AzureBlobStorage::getAuthMethod(config, config_prefix),
            .client_options = AzureBlobStorage::getClientOptions(context, context->getSettingsRef(), *azure_settings, /*for_disk=*/ true),
        };

        return std::make_shared<AzureObjectStorage>(
            name,
            params.auth_method, AzureBlobStorage::getContainerClient(params, /*readonly=*/ false), std::move(azure_settings),
            params, params.endpoint.prefix.empty() ? params.endpoint.container_name : params.endpoint.container_name + "/" + params.endpoint.prefix,
            params.endpoint.getServiceEndpoint(), common_key_prefix);
    };

    factory.registerObjectStorageType("azure_blob_storage", creator);
    factory.registerObjectStorageType("azure", creator);
    factory.registerObjectStorageType("azure_plain", creator);
    factory.registerObjectStorageType("azure_plain_rewritable", creator);
}
#endif

void registerWebObjectStorage(ObjectStorageFactory & factory)
{
    factory.registerObjectStorageType("web", [](
        const std::string & /* name */,
        const Poco::Util::AbstractConfiguration & config,
        const std::string & config_prefix,
        const ContextPtr & context,
        bool /* skip_access_check */) -> ObjectStoragePtr
    {
        auto uri = context->getMacros()->expand(config.getString(config_prefix + ".endpoint"));
        if (!uri.ends_with('/'))
            throw Exception(
                ErrorCodes::BAD_ARGUMENTS, "URI must end with '/', but '{}' doesn't.", uri);
        try
        {
            Poco::URI poco_uri(uri);
        }
        catch (const Poco::Exception & e)
        {
            throw Exception(
                ErrorCodes::BAD_ARGUMENTS, "Bad URI: `{}`. Error: {}", uri, e.what());
        }

        return std::make_shared<WebObjectStorage>(uri, context);
    });
}

void registerLocalObjectStorage(ObjectStorageFactory & factory)
{
    auto creator = [](
        const std::string & name,
        const Poco::Util::AbstractConfiguration & config,
        const std::string & config_prefix,
        const ContextPtr & context,
        bool /* skip_access_check */) -> ObjectStoragePtr
    {
        String object_key_prefix;
        UInt64 keep_free_space_bytes;
        loadDiskLocalConfig(name, config, config_prefix, context, object_key_prefix, keep_free_space_bytes);

        /// keys are mapped to the fs, object_key_prefix is a directory also
        fs::create_directories(object_key_prefix);

        bool read_only = config.getBool(config_prefix + ".readonly", false);
        LocalObjectStorageSettings settings(object_key_prefix, read_only);

        return std::make_shared<LocalObjectStorage>(settings);
    };

    factory.registerObjectStorageType("local_blob_storage", creator);
    factory.registerObjectStorageType("local", creator);
    factory.registerObjectStorageType("local_plain", creator);
    factory.registerObjectStorageType("local_plain_rewritable", creator);
}

void registerObjectStorages()
{
    auto & factory = ObjectStorageFactory::instance();

#if USE_AWS_S3
    registerS3ObjectStorage(factory);
#endif

#if USE_HDFS
    registerHDFSObjectStorage(factory);
#endif

#if USE_AZURE_BLOB_STORAGE
    registerAzureObjectStorage(factory);
#endif

    registerWebObjectStorage(factory);
    registerLocalObjectStorage(factory);
}

void ObjectStorageFactory::clearRegistry()
{
    registry.clear();
}
}
